package com.mask.im.sdk.listener;

import com.alibaba.fastjson.JSON;
import com.mask.im.sdk.annotation.RedisMsgListener;
import com.mask.im.sdk.dto.IMPushMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Redis消息监听器注解处理器
 * 扫描@RedisMsgListener注解的方法并自动消费消息
 *
 * @author jian
 */
@Slf4j
@Component
public class RedisMsgListenerProcessor {

    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Value("${redis.msg.default-queue:im:msg:queue}")
    private String defaultQueue;

    @Value("${im.service.node-id:default-node}")
    private String currentNodeId;


    // 动态节点ID获取器（可选）
    private java.util.function.Supplier<String> dynamicNodeIdSupplier;

    // 存储监听器方法
    private final Map<String, List<ListenerMethod>> listenerMethods = new ConcurrentHashMap<>();

    // 线程池
    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
    
    // 已启动的消费者线程（用于避免重复启动）
    private final Set<String> startedConsumers = ConcurrentHashMap.newKeySet();
    
    // 初始化标志
    private volatile boolean initialized = false;

    @PostConstruct
    public void init() {
        log.info("开始初始化Redis消息监听器注解处理器...");
        
        // 自动查找并设置动态节点ID获取器（如果存在ServiceIdManager）
        // 注意：这里只设置获取器，不重新扫描（避免重复）
        tryAutoConfigureNodeIdSupplier(false);
        
        scanListenerMethods();
        startConsumers();
        initialized = true;
        log.info("Redis消息监听器注解处理器已启动，发现 {} 个监听器", getTotalListenerCount());
        log.info("监听的队列: {}", listenerMethods.keySet());
        log.info("当前节点ID: {}", getCurrentNodeId());
    }
    
    /**
     * 自动查找并配置动态节点ID获取器
     * 如果Spring容器中存在ServiceIdManager bean，自动设置动态节点ID获取器
     * 
     * @param rescan 是否重新扫描监听器（初始化时为false，避免重复）
     */
    private void tryAutoConfigureNodeIdSupplier(boolean rescan) {
        try {
            // 尝试通过反射查找ServiceIdManager（避免直接依赖server模块）
            String[] beanNames = applicationContext.getBeanNamesForType(
                java.lang.Class.forName("com.mask.im.server.service.ServiceIdManager"));
            
            if (beanNames != null && beanNames.length > 0) {
                Object serviceIdManager = applicationContext.getBean(beanNames[0]);
                // 通过反射调用getNodeId()方法
                Method getNodeIdMethod = serviceIdManager.getClass().getMethod("getNodeId");
                setDynamicNodeIdSupplier(() -> {
                    try {
                        return (String) getNodeIdMethod.invoke(serviceIdManager);
                    } catch (Exception e) {
                        log.error("动态获取节点ID失败", e);
                        return currentNodeId;
                    }
                }, rescan);
                log.info("已自动配置动态节点ID获取器（通过ServiceIdManager）");
            } else {
                log.info("未找到ServiceIdManager bean，使用配置的节点ID: {}", currentNodeId);
            }
        } catch (ClassNotFoundException e) {
            // ServiceIdManager类不存在，说明不在server模块中，使用配置的节点ID
            log.debug("ServiceIdManager类不存在，使用配置的节点ID: {}", currentNodeId);
        } catch (Exception e) {
            log.warn("自动配置动态节点ID获取器失败，使用配置的节点ID: {}", currentNodeId, e);
        }
    }


    /**
     * 扫描所有@RedisMsgListener注解的方法
     */
    private void scanListenerMethods() {
        String[] beanNames = applicationContext.getBeanDefinitionNames();

        for (String beanName : beanNames) {
            Object bean = applicationContext.getBean(beanName);
            Class<?> beanClass = bean.getClass();
            // 获取所有方法
            Method[] methods = beanClass.getDeclaredMethods();
            for (Method method : methods) {
                RedisMsgListener annotation = method.getAnnotation(RedisMsgListener.class);
                if (annotation != null) {
                    System.err.println("Bean名称: " + beanName + ", 类型: " + applicationContext.getBean(beanName).getClass());
                    registerListenerMethod(bean, method, annotation);
                }
            }
        }
    }

    /**
     * 注册监听器方法
     */
    private void registerListenerMethod(Object bean, Method method, RedisMsgListener annotation) {
        String queueName = determineQueueName(annotation);
        ListenerMethod listenerMethod = new ListenerMethod();
        listenerMethod.setBean(bean);
        listenerMethod.setMethod(method);
        listenerMethod.setQueueName(queueName);
        listenerMethod.setMessageTypes(Arrays.asList(annotation.messageTypes()));
        listenerMethod.setBatchSize(annotation.batchSize());
        listenerMethods.computeIfAbsent(queueName, k -> new ArrayList<>()).add(listenerMethod);
        log.info("注册监听器: {}#{} -> 队列: {}, 消息类型: {}",
                bean.getClass().getSimpleName(), method.getName(), queueName,
                Arrays.toString(annotation.messageTypes()));
    }

    /**
     * 确定队列名称
     * 优先级: nodeId > queue > 当前节点队列
     */
    private String determineQueueName(RedisMsgListener annotation) {
        // 1. 如果指定了nodeId，使用nodeId构建队列名
        if (!annotation.nodeId().isEmpty()) {
            String nodeId = annotation.nodeId();
            // 特殊处理：如果是"CURRENT_NODE"，动态获取当前节点ID
            if ("CURRENT_NODE".equals(nodeId)) {
                nodeId = getCurrentNodeId();
            }
            return "im:msg:queue:" + nodeId;
        }

        //  如果指定了queue，使用指定的队列名
        if (!annotation.queue().isEmpty()) {
            return annotation.queue();
        }

        // 默认使用当前节点的队列
        return "im:msg:queue:" + getCurrentNodeId();
    }

    /**
     * 获取当前节点ID
     * 优先使用动态获取器，否则使用配置的节点ID
     */
    private String getCurrentNodeId() {
        if (dynamicNodeIdSupplier != null) {
            try {
                return dynamicNodeIdSupplier.get();
            } catch (Exception e) {
                log.warn("动态获取节点ID失败，使用配置的节点ID: {}", e.getMessage());
            }
        }
        return currentNodeId;
    }

    /**
     * 设置动态节点ID获取器
     * 用于在运行时动态获取节点ID
     * 
     * @param supplier 节点ID获取器
     * @param rescan 是否重新扫描监听器（初始化时为false）
     */
    public void setDynamicNodeIdSupplier(java.util.function.Supplier<String> supplier, boolean rescan) {
        this.dynamicNodeIdSupplier = supplier;
        log.info("动态节点ID获取器已设置");
        // 如果已经初始化过，且需要重新扫描，则重新扫描监听器方法
        if (supplier != null && rescan && initialized) {
            log.info("重新扫描监听器方法...");
            // 清空旧的监听器
            listenerMethods.clear();
            startedConsumers.clear();
            scanListenerMethods();
            startConsumers();
            log.info("监听器重新扫描完成，发现 {} 个监听器", getTotalListenerCount());
            log.info("监听的队列: {}", listenerMethods.keySet());
        }
    }
    
    /**
     * 设置动态节点ID获取器（兼容旧方法，默认重新扫描）
     */
    public void setDynamicNodeIdSupplier(java.util.function.Supplier<String> supplier) {
        setDynamicNodeIdSupplier(supplier, true);
    }

    /**
     * 启动消费者
     */
    private void startConsumers() {
        // 启动所有已注册监听器的队列的消费者（避免重复启动）
        for (String queueName : listenerMethods.keySet()) {
            if (!startedConsumers.contains(queueName)) {
                executorService.submit(() -> consumeMessages(queueName));
                startedConsumers.add(queueName);
                log.info("启动队列消费者: {}", queueName);
            } else {
                log.debug("队列 {} 的消费者已启动，跳过", queueName);
            }
        }
        log.info("已启动 {} 个队列的消费者: {}", startedConsumers.size(), startedConsumers);
    }


    /**
     * 消费消息
     */
    private void consumeMessages(String queueName) {
        log.info("开始消费队列: {}", queueName);
        boolean keyTypeChecked = false;
        while (!Thread.currentThread().isInterrupted()) {
            try {
                // 只在第一次检查key类型，避免重复检查
                if (!keyTypeChecked) {
                    ensureKeyIsList(queueName);
                    keyTypeChecked = true;
                }

                // 阻塞5秒等待消息
                Object message = redisTemplate.opsForList().rightPop(queueName, 5, TimeUnit.SECONDS);
                if (message != null) {
                    log.info("从队列 {} 收到消息: {}", queueName, message);
                    IMPushMessage msg = JSON.parseObject(message.toString(), IMPushMessage.class);
                    handleMessage (queueName, msg);
                }
            } catch (Exception e) {
                // 如果是WRONGTYPE错误，重新检查key类型
                if (e.getMessage() != null && e.getMessage().contains("WRONGTYPE")) {
                    log.warn("队列 {} 类型错误，重新检查", queueName);
                    ensureKeyIsList(queueName);
                    keyTypeChecked = true;
                } else {
                    log.error("消费消息异常，队列: {}", queueName, e);
                }
                sleep(1000);
            }
        }
    }

    /**
     * 处理消息
     */
    private void handleMessage(String queueName, IMPushMessage msg) {
        // 获取当前队列的监听器
        List<ListenerMethod> methods = listenerMethods.get(queueName);
        // 如果没有当前队列的监听器，尝试使用默认队列的监听器
        if (methods == null || methods.isEmpty()) {
            methods = listenerMethods.get(defaultQueue);
        }
        // 如果还是没有，尝试使用所有监听器
        if (methods == null || methods.isEmpty()) {
            methods = new ArrayList<>();
            for (List<ListenerMethod> queueMethods : listenerMethods.values()) {
                methods.addAll(queueMethods);
            }
        }


        if (methods.isEmpty()) {
            log.warn("队列 {} 没有可用的监听器", queueName);
            return;
        }

        for (ListenerMethod listenerMethod : methods) {
            try {
                // 检查消息类型是否匹配
                if (isMessageTypeMatch(msg.getPushType(), listenerMethod.getMessageTypes())) {
                    // 调用监听器方法
                    listenerMethod.getMethod().invoke(listenerMethod.getBean(), msg);
                    log.info("消息已处理: {} -> {} (队列: {})", msg.getSenderId(), listenerMethod.getMethod().getName(), queueName);
                }
            } catch (Exception e) {
                log.error("调用监听器方法失败: {}#{}",
                        listenerMethod.getBean().getClass().getSimpleName(),
                        listenerMethod.getMethod().getName(), e);
            }
        }
    }

    /**
     * 检查消息类型是否匹配
     */
    private boolean isMessageTypeMatch(String pushType, List<String> targetTypes) {
        if (targetTypes.contains("*")) {
            return true;
        }
        return targetTypes.contains(pushType);
    }

    /**
     * 确保key是List类型
     */
    private void ensureKeyIsList(String queueName) {
        try {
            String keyType = redisTemplate.type(queueName).code();
            if (keyType != null && !"list".equals(keyType.toLowerCase()) && !"none".equals(keyType.toLowerCase())) {
                log.warn("队列key {} 的类型是 {}，不是List类型，将删除重建", queueName, keyType);
                redisTemplate.delete(queueName);
            }
            // 如果key不存在(none)或已经是list类型，不输出日志
        } catch (Exception e) {
            log.warn("检查队列key {} 类型时出错: {}", queueName, e.getMessage());
            try {
                redisTemplate.delete(queueName);
            } catch (Exception deleteException) {
                log.error("删除队列key {} 失败: {}", queueName, deleteException.getMessage());
            }
        }
    }

    /**
     * 获取监听器总数
     */
    private int getTotalListenerCount() {
        return listenerMethods.values().stream().mapToInt(List::size).sum();
    }

    /**
     * 睡眠
     */
    private void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @PreDestroy
    public void destroy() {
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
        log.info("Redis消息监听器注解处理器已停止");
    }

    /**
     * 监听器方法信息
     */
    private static class ListenerMethod {
        private Object bean;
        private Method method;
        private String queueName;
        private List<String> messageTypes;
        private int batchSize;

        // Getters and Setters
        public Object getBean() {
            return bean;
        }

        public void setBean(Object bean) {
            this.bean = bean;
        }

        public Method getMethod() {
            return method;
        }

        public void setMethod(Method method) {
            this.method = method;
        }

        public String getQueueName() {
            return queueName;
        }

        public void setQueueName(String queueName) {
            this.queueName = queueName;
        }

        public List<String> getMessageTypes() {
            return messageTypes;
        }

        public void setMessageTypes(List<String> messageTypes) {
            this.messageTypes = messageTypes;
        }

        public int getBatchSize() {
            return batchSize;
        }

        public void setBatchSize(int batchSize) {
            this.batchSize = batchSize;
        }
    }
}
